package top.scauwlt.remoting.transport.codec;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import lombok.extern.slf4j.Slf4j;
import top.scauwlt.compress.Compress;
import top.scauwlt.config.RpcConfig;
import top.scauwlt.enums.CompressTypeEnum;
import top.scauwlt.enums.SerializationTypeEnum;
import top.scauwlt.extension.ExtensionLoader;
import top.scauwlt.remoting.constants.RpcConstants;
import top.scauwlt.remoting.dto.RpcMessage;
import top.scauwlt.remoting.dto.RpcRequest;
import top.scauwlt.remoting.dto.RpcResponse;
import top.scauwlt.serialize.Serializer;

/**
 * 消息解码器
 *
 * @Author wmm
 * @Date 2022/7/26 14:55
 */
@Slf4j
public class RpcMessageDecoder extends LengthFieldBasedFrameDecoder {
    /**
     * 各参数见RpcConstants
     */
    public RpcMessageDecoder() {
        this(RpcConstants.MAX_FRAME_LENGTH, RpcConstants.LENGTH_FIELD_OFFSET,
                RpcConstants.LENGTH_FILED_LENGTH, RpcConstants.LENGTH_ADJUSTMENT,
                RpcConstants.INITIAL_BYTES_TO_STRIP);
    }

    public RpcMessageDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment, int initialBytesToStrip) {
        super(maxFrameLength, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip);
    }

    @Override
    protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        Object decode = super.decode(ctx, in);
        if (decode instanceof ByteBuf) {
            ByteBuf frame = (ByteBuf) decode;
            if (frame.readableBytes() >= RpcConstants.HEAD_LENGTH) {
                try {
                    return decodeFrame(frame);
                } catch (Exception e) {
                    log.error("Decode message to RpcMessage fail : {}", e.getMessage());
                    throw e;
                } finally {
                    frame.release();
                }
            }
        }
        return decode;
    }

    private Object decodeFrame(ByteBuf in) {
        checkMagicCode(in);
        checkVersion(in);
        int fullLength = in.readInt();

        byte messageType = in.readByte();
        byte codecType = in.readByte();
        byte compressType = in.readByte();
        int requestId = in.readInt();
        RpcMessage rpcMessage = RpcMessage.builder()
                .messageType(messageType)
                .codec(codecType)
                .compress(compressType)
                .requestId(requestId)
                .build();

        if (messageType == RpcConstants.HEARTBEAT_REQUEST_TYPE) {
            rpcMessage.setData(RpcConstants.PING);
            return rpcMessage;
        }
        if (messageType == RpcConstants.HEARTBEAT_RESPONSE_TYPE) {
            rpcMessage.setData(RpcConstants.PONG);
            return rpcMessage;
        }

        //获取到消息数据体字节数组（压缩、序列化状态）
        int bodyLength = fullLength - RpcConstants.HEAD_LENGTH;
        if (bodyLength > 0) {
            byte[] bodyBytes = new byte[bodyLength];
            in.readBytes(bodyBytes);

            //解压缩
            String compressName = CompressTypeEnum.getName(compressType);
            Compress compress = ExtensionLoader.getExtensionLoader(Compress.class).getExtension(compressName);
            bodyBytes = compress.decompress(bodyBytes);

            //反序列化
            String serializationName = SerializationTypeEnum.getName(codecType);
            Serializer serializer = ExtensionLoader.getExtensionLoader(Serializer.class).getExtension(serializationName);

            if (messageType == RpcConstants.REQUEST_TYPE) {
                RpcRequest data = serializer.deserialize(bodyBytes, RpcRequest.class);
                rpcMessage.setData(data);
            } else if (messageType == RpcConstants.RESPONSE_TYPE) {
                RpcResponse data = serializer.deserialize(bodyBytes, RpcResponse.class);
                rpcMessage.setData(data);
            }
        }
        return rpcMessage;
    }

    private void checkMagicCode(ByteBuf in) {
        int len = RpcConstants.MAGIC_CODE.length;
        byte[] code = new byte[len];
        in.readBytes(code);
        for (int i = 0; i < len; i++) {
            if (RpcConstants.MAGIC_CODE[i] != code[i]) {
                throw new IllegalArgumentException("Unknown magic code: " + new String(code));
            }
        }
    }

    private void checkVersion(ByteBuf in) {
        byte version = in.readByte();
        if (version != RpcConstants.VERSION) {
            throw new RuntimeException("version isn't compatible");
        }
    }
}
